package com.gaussdb.tansform;

import com.gaussdb.tool.JsonArrayUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple5;
import org.json.JSONObject;

public class SourceMapFunction implements MapFunction<String, Tuple5<String,String, String, String, String>> {

    @Override
    public Tuple5<String, String,String, String, String> map(String value) throws Exception {
        // 生成json对象
        JSONObject jsonObject = new JSONObject(value);

        // 表名
        String tableName = jsonObject.getString("table_name");

        // 操作类型
        String opType = jsonObject.getString("op_type");

        // 字段名称
        String columnsNameStr=null;
        // 字段类型
        String columnsTypeStr=null;
        // 字段内容
        String columnsValStr=null;

        if (opType.equals("DELETE")){
            // 字段名称
            columnsNameStr = JsonArrayUtil.JsonArrayToString(jsonObject, "old_keys_name");
            // 字段类型
            columnsTypeStr = JsonArrayUtil.JsonArrayToString(jsonObject, "old_keys_type");
            // 字段内容
            columnsValStr = JsonArrayUtil.JsonArrayToString(jsonObject, "old_keys_val");
        }else {
            // 字段名称
            columnsNameStr = JsonArrayUtil.JsonArrayToString(jsonObject, "columns_name");
            // 字段类型
            columnsTypeStr = JsonArrayUtil.JsonArrayToString(jsonObject, "columns_type");
            // 字段内容
            columnsValStr = JsonArrayUtil.JsonArrayToString(jsonObject, "columns_val");
        }


        // 返回包含表名、字段名称、字段类型和字段内容的元组
        return Tuple5.of(tableName, opType,columnsNameStr, columnsTypeStr, columnsValStr);
    }
}
